package com.ljm.reactor.Sinks;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;//必须指明  reactor-core 版本 3.4.0-SNAPSHOT)
import reactor.core.scheduler.Schedulers;

public class ReplayLimitExample {
    public static void main(String[] args) {

        /*
        一、代码解释
            1、创建Sink: 使用 Sinks.many().replay().limit(3) 创建一个多播的重放 Sink，并限制其重放缓存为最近的 3 条消息。
            2、创建Flux: 将 Sink 转换为一个 Flux，并使用 publishOn(Schedulers.parallel()) 来在并行调度器上发布消息。
            3、订阅者1: 首先订阅 Sink，并接收所有发布的消息。
            4、发布消息: 发布 5 条消息到 Sink。
            5、延时: 让主线程等待一段时间，以确保所有消息都已发布并被第一个订阅者接收到。
            6、订阅者2: 在稍后订阅 Sink，并接收最近的 3 条消息（“Message 3”, “Message 4”, “Message 5”）。
            7、发布更多消息: 继续发布更多的消息到 Sink。
            8、等待: 再次让主线程等待，以确保所有消息都已处理完毕
        */
        try {
            // 创建一个Sinks.Many对象，使用replay()和limit()来限制重放的消息数量
            Sinks.Many<String> sink = Sinks.many().replay().limit(3);

            // 创建一个Flux，将从sink接收数据
            Flux<String> flux = sink.asFlux().publishOn(Schedulers.parallel());

            // 订阅者1
            flux.subscribe(data -> System.out.println("Subscriber 1 received: " + data));

            // 发布一些数据
            sink.tryEmitNext("Message 1");
            sink.tryEmitNext("Message 2");
            sink.tryEmitNext("Message 3");
            sink.tryEmitNext("Message 4");
            sink.tryEmitNext("Message 5");

            // 让主线程等待一段时间，以确保上述数据都已发布
            Thread.sleep(1000);

            // 订阅者2在稍后订阅，将会接收到最近的3条消息
            flux.subscribe(data -> System.out.println("Subscriber 2 received: " + data));

            // 发布更多数据
            sink.tryEmitNext("Message 6");
            sink.tryEmitNext("Message 7");

            // 让主线程等待，以确保所有消息都已处理完毕
            Thread.sleep(1000);
        }catch(Exception ex){

        }

    }
}
